﻿#include "RpcEndpoint.h"
#include "XyNet/Channel/RpcSocketChannel.h"
#include "XyNet/Common/Exception.h"
#include "XyNet/Common/Logger.h"
#include "XyNet/Common/StringConverter.h"
#include "XyNet/Protocol/cpp/Rpc.service.h"
#include "XyNet/Service/RpcController.h"
#include "XyNet/Service/RpcServiceManager.h"

namespace XyNet
{
	RpcEndpoint::Config::Config(const string& baseFileName /*= ""*/) : InfoConfig(baseFileName) {}

	//////////////////////////////////////////////////////////////////////////

	XYNET_IMPLEMENT_DEFAULT_SUBJECT(RpcEndpointObserver);

	RpcEndpoint::RpcEndpoint(const string& name, shared_ptr<Config> config)
		: _name(name), _config(config)
		, _ioService(/*boost::thread::hardware_concurrency() * 2*/)
		, _serviceManager(make_shared<RpcServiceManager>())
	{}

	RpcEndpoint::~RpcEndpoint()
	{
		stop();
	}

	shared_ptr<const RpcEndpoint::Config> RpcEndpoint::getConfig() const
	{
		return _config;
	}

	boost::asio::io_service* RpcEndpoint::getIoService()
	{
		return &_ioService;
	}

	shared_ptr<RpcServiceManager> RpcEndpoint::getServiceManager()
	{
		return _serviceManager;
	}

	shared_ptr<RpcSocketChannel> RpcEndpoint::createSocketChannel(
		shared_ptr<boost::asio::ip::tcp::socket> socket)
	{
		return make_shared<RpcSocketChannel>(this, socket);
	}

	void RpcEndpoint::handleSocketException(
		shared_ptr<boost::asio::ip::tcp::socket> socket, const boost::system::error_code& error, shared_ptr<RpcSocketChannel> socketChannel)
	{
		if (getIoService()->stopped())
			return;

		// 获取远端地址，生成消息字符串
		boost::system::error_code ec;
		auto remoteEndpoint = socket->remote_endpoint(ec);

		boost::format format("%s:%s %s: %s");
		if (!ec)
			format % remoteEndpoint.address() % remoteEndpoint.port();
		else
			format % "Unknown_address" % "Unknown_port";
		format % error.category().name() % error.message();

		if (socketChannel)
			handleSocketChannelException(socketChannel, RpcErrors::SocketError);

		// 输出消息
		XYNET_HANDLE_EXCEPTION(Exception(StringConverter::localToUtf8(format.str())));
	}

	void RpcEndpoint::handleSocketChannelException(shared_ptr<RpcSocketChannel> socketChannel, int errorCode)
	{
		notify(bind(&RpcEndpointObserver::onSocketChannelException, 
			_1, socketChannel, errorCode));
	}

	void RpcEndpoint::blockRun()
	{
		boost::asio::signal_set signals(*getIoService());
		signals.add(SIGINT);
		signals.add(SIGTERM);
#ifdef SIGBREAK
		signals.add(SIGBREAK);
#endif
#ifdef SIGQUIT
		signals.add(SIGQUIT);
#endif
		signals.async_wait(bind(&boost::asio::io_service::stop, &_ioService));

		//boost::asio::io_service::work work(_ioService);
		_ioService.run();
	}

	void RpcEndpoint::run(int concurrency)
	{
		assert(_ioServiceThreads.empty());

		Logger::getDefault()->log(_name + " running.");

		boost::asio::io_service::work work(_ioService);
		for (int i = 0; i < concurrency; ++i)
		{
			_ioServiceThreads.emplace_back(make_shared<thread>([this]()
			{
#ifndef _DEBUG
				try
				{
#endif // _DEBUG
					_ioService.run();
#ifndef _DEBUG
				}
				catch (const exception& e)
				{
					XYNET_HANDLE_EXCEPTION(e);
				}
#endif // _DEBUG
			}));
		}
	}

	void RpcEndpoint::stop()
	{
		//assert(!_ioServiceThreads.empty());
		if (_ioServiceThreads.empty())
			return;

		Logger::getDefault()->log(_name + " stopping.");

		_ioService.stop();

		for (size_t i = 0; i < _ioServiceThreads.size(); ++i)
		{
			_ioServiceThreads[i]->join();
			_ioServiceThreads[i].reset();
		}
		_ioServiceThreads.clear();

		_ioService.reset();
	}
}